package com.zlsf.publish;


import com.zlsf.codec.AudioCodec;
import com.zlsf.entity.Media;
import com.zlsf.entity.MediaEncoding;
import com.zlsf.flv.FlvEncoder;
import com.zlsf.util.ByteHolder;
import io.vertx.ext.web.RoutingContext;
import lombok.extern.slf4j.Slf4j;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

@Slf4j
public class Channel {

  ConcurrentLinkedQueue<Subscriber> subscribers;

  String tag;
  boolean publishing;
  ByteHolder buffer;
  AudioCodec audioCodec;
  FlvEncoder flvEncoder;
  private long firstTimestamp = -1;

  private int zeroCount; // 记录连续0次的计数

  public Channel(String tag) {
    this.tag = tag;
    this.subscribers = new ConcurrentLinkedQueue<Subscriber>();
    this.flvEncoder = new FlvEncoder(true, true);
    this.buffer = new ByteHolder(2048 * 100);

  }

  public boolean isPublishing() {
    return publishing;
  }

  public Subscriber subscribe(RoutingContext ctx) {
    log.info("channel: {} -> {}, subscriber: {}", Long.toHexString(hashCode() & 0xffffffffL), tag, ctx.request().remoteAddress());

    Subscriber subscriber = new VideoSubscriber(this.tag, ctx);
    this.subscribers.add(subscriber);
    return subscriber;
  }

  public void writeAudio(long timestamp, int pt, byte[] data) {
    if (audioCodec == null) {
      audioCodec = AudioCodec.getCodec(pt);
      log.info("audio codec: {}", MediaEncoding.getEncoding(Media.Type.Audio, pt));
    }
    broadcastAudio(timestamp, audioCodec.toPCM(data));
  }


  public void writeVideo(long sequence, long timeoffset, int payloadType, byte[] h264) {
    if (firstTimestamp == -1) firstTimestamp = timeoffset;
    this.publishing = true;
    this.buffer.write(h264);
    while (true) {
      byte[] nalu = readNalu();
      if (nalu == null) break;
      if (nalu.length < 4) continue;

      byte[] flvTag = this.flvEncoder.write(nalu, (int) (timeoffset - firstTimestamp));
      if (flvTag == null) continue;

      // 广播给所有的观众
      broadcastVideo(timeoffset, flvTag);
    }
  }

  public void broadcastVideo(long timeoffset, byte[] flvTag) {
    for (Subscriber subscriber : subscribers) {
      subscriber.onVideoData(timeoffset, flvTag, flvEncoder);
    }
  }

  public void broadcastAudio(long timeoffset, byte[] flvTag) {
    for (Subscriber subscriber : subscribers) {
      subscriber.onAudioData(timeoffset, flvTag, flvEncoder);
    }
  }

  public void unsubscribe(RoutingContext ctx) {
    for (Iterator<Subscriber> itr = subscribers.iterator(); itr.hasNext(); ) {
      Subscriber subscriber = itr.next();
      if (subscriber.getContext().hashCode() == ctx.hashCode()) {
        itr.remove();
        subscriber.close();
        return;
      }
    }
  }

  public void close() {
    for (Iterator<Subscriber> itr = subscribers.iterator(); itr.hasNext(); ) {
      Subscriber subscriber = itr.next();
      subscriber.close();
      itr.remove();
    }
  }

  private byte[] readNalu() {
    for (int i = 0; i < buffer.size() - 3; i++) {
      int a = buffer.get(i + 0) & 0xff;
      int b = buffer.get(i + 1) & 0xff;
      int c = buffer.get(i + 2) & 0xff;
      int d = buffer.get(i + 3) & 0xff;
      if (a == 0x00 && b == 0x00 && c == 0x00 && d == 0x01) {
        if (i == 0) continue;
        byte[] nalu = new byte[i];
        buffer.sliceInto(nalu, i);
        return nalu;
      }
    }
    return null;
  }

  public int getSubscriberCount() {
    return subscribers.size();
  }

  public int getZeroCount() {
    return zeroCount;
  }

  public void incrementZeroCount() {
    zeroCount++;
  }

  public void resetZeroCount() {
    zeroCount = 0;
  }
}
